#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2024/5/11 23:18
# @Author  : YueJian
# @File    : tools.py
# @Description : 工具类集合

import hashlib
import importlib
import pkgutil
import re
import string
import datetime
import random
import time
import uuid
from typing import List, Union, Dict, Any, Optional, TypeVar, Generic, Callable, Pattern
from functools import lru_cache
from XdbSearchIP.xdbSearcher import XdbSearcher
from fastapi import Request
from user_agents import parse

from applications import settings
from applications.log import logger

# 类型定义
T = TypeVar("T")
DictT = TypeVar("DictT", bound=Dict[str, Any])
ListT = TypeVar("ListT", bound=List[Any])

# 常量定义
SYMBOLS = {
    "customary": ("B", "K", "M", "G", "T", "P", "E", "Z", "Y"),
    "customary_ext": ("byte", "kilo", "mega", "giga", "tera", "peta", "exa", "zetta", "iotta"),
    "iec": ("Bi", "Ki", "Mi", "Gi", "Ti", "Pi", "Ei", "Zi", "Yi"),
    "iec_ext": ("byte", "kibi", "mebi", "gibi", "tebi", "pebi", "exbi", "zebi", "yobi"),
}

# 正则表达式预编译
CAMEL_CASE_PATTERN: Pattern = re.compile(r"([a-z0-9])([A-Z])")
SNAKE_CASE_PATTERN: Pattern = re.compile(r"([a-z0-9])([A-Z])")


class ToolsUtils:
    """工具类集合，提供各种通用功能"""

    @staticmethod
    def random_string(length: int = 10) -> str:
        """生成指定长度的随机字符串

        Args:
            length: 字符串长度，默认10

        Returns:
            str: 随机字符串
        """
        return "".join(random.choices(string.ascii_letters + string.digits, k=length))

    @staticmethod
    def build_tree(menu: List[DictT], parent_id: Optional[Any] = None, parent_name: str = "parentId") -> List[DictT]:
        """构建树形结构

        Args:
            menu: 菜单列表
            parent_id: 父节点ID
            parent_name: 父节点字段名

        Returns:
            List[DictT]: 树形结构
        """
        tree = []
        for item in menu:
            node = item.copy()
            if node.get(parent_name) == parent_id:
                children = ToolsUtils.build_tree(menu, node.get("id"), parent_name)
                if children:
                    node["children"] = children
                tree.append(node)
        return tree

    @staticmethod
    def options_tree(
        menu: List[DictT],
        parent_id: Optional[Any] = None,
        parent_name: str = "parentId",
        label: str = "name",
        value: str = "id",
    ) -> List[Dict[str, Any]]:
        """构建选项树形结构

        Args:
            menu: 菜单列表
            parent_id: 父节点ID
            parent_name: 父节点字段名
            label: 标签字段名
            value: 值字段名

        Returns:
            List[Dict[str, Any]]: 选项树形结构
        """
        tree = []
        for item in menu:
            if item.get(parent_name) == parent_id:
                node = {
                    "label": item.get(label),
                    "value": item.get(value),
                }
                children = ToolsUtils.options_tree(menu, item.get(value), parent_name, label, value)
                if children:
                    node["children"] = children
                tree.append(node)
        return tree

    @staticmethod
    def list_dict_find(options: List[DictT], key: str, value: Any) -> Optional[DictT]:
        """在字典列表中查找指定键值的项

        Args:
            options: 字典列表
            key: 键名
            value: 值

        Returns:
            Optional[DictT]: 找到的项，未找到返回None
        """
        return next((item for item in options if item.get(key) == value), None)

    @staticmethod
    def get_time_interval(start_time: str, end_time: str, interval: int, time_format: str = "%H:%M:%S") -> List[str]:
        """获取时间间隔列表

        Args:
            start_time: 开始时间
            end_time: 结束时间
            interval: 间隔时间（分）
            time_format: 时间格式

        Returns:
            List[str]: 时间间隔列表
        """
        if start_time.count(":") == 1:
            start_time = f"{start_time}:00"
        if end_time.count(":") == 1:
            end_time = f"{end_time}:00"

        start_time = datetime.datetime.strptime(start_time, "%H:%M:%S")
        end_time = datetime.datetime.strptime(end_time, "%H:%M:%S")

        time_range = []
        while end_time > start_time:
            time_range.append(start_time.strftime(time_format))
            start_time = start_time + datetime.timedelta(minutes=interval)
        return time_range

    @staticmethod
    def generate_string(length: int = 8) -> str:
        """生成随机字符串

        Args:
            length: 字符串长度

        Returns:
            str: 随机字符串
        """
        return "".join(random.sample(string.ascii_letters + string.digits, length))

    @staticmethod
    def is_private_ip(ip: str) -> bool:
        """判断是否为私有IP

        Args:
            ip: IP地址

        Returns:
            bool: 是否为私有IP
        """
        if ip == "127.0.0.1":
            return True

        try:
            ip_parts = ip.split(".")
            if len(ip_parts) != 4:
                return False

            first_octet = int(ip_parts[0])
            second_octet = int(ip_parts[1])

            return (
                first_octet == 10
                or (first_octet == 172 and 16 <= second_octet <= 31)
                or (first_octet == 192 and second_octet == 168)
            )
        except (ValueError, IndexError):
            return False

    @staticmethod
    def get_ip_address(request: Request) -> str:
        """获取请求的IP地址

        Args:
            request: FastAPI请求对象

        Returns:
            str: IP地址
        """
        x_forwarded_for = request.headers.get("x-forwarded-for", "")
        if x_forwarded_for:
            return x_forwarded_for.split(",")[0].strip()

        x_real_ip = request.headers.get("X-Real-Ip", "")
        if x_real_ip:
            return x_real_ip

        return request.client.host

    @staticmethod
    async def get_ip_reality_address(ip: str) -> Dict[str, Any]:
        """获取IP地址的真实地理位置

        Args:
            ip: IP地址

        Returns:
            Dict[str, Any]: 地理位置信息
        """
        import httpx

        if ToolsUtils.is_private_ip(ip):
            return {
                "continent": "内网",
                "country": "内网",
                "province": "内网",
                "city": "内网",
                "district": "内网",
                "isp": "内网",
                "latitude": None,
                "longitude": None,
                "area_code": None,
                "city_code": None,
            }

        try:
            url = f"https://api.qjqq.cn/api/district?ip={ip}"
            async with httpx.AsyncClient(verify=False, timeout=5.0) as client:
                response = await client.get(url)
                data = response.json()

                if data["code"] == 200:
                    return {
                        "continent": data["data"]["continent"],
                        "country": data["data"]["country"],
                        "province": data["data"]["prov"],
                        "city": data["data"]["city"],
                        "district": data["data"]["district"],
                        "isp": data["data"]["isp"],
                        "latitude": data["data"]["lat"],
                        "longitude": data["data"]["lng"],
                        "area_code": data["data"]["area_code"],
                        "city_code": data["data"]["city_code"],
                    }
                return {}
        except Exception as e:
            logger.error(f"IP地址解析失败: {str(e)}")
            return {}

    @staticmethod
    def get_agent_info(request: Request) -> Dict[str, str]:
        """获取用户代理信息

        Args:
            request: FastAPI请求对象

        Returns:
            Dict[str, str]: 用户代理信息
        """
        try:
            user_agent = parse(request.headers.get("user-agent", ""))
            return {
                "os": f"{user_agent.os.family} {user_agent.os.version_string}",
                "browser": f"{user_agent.browser.family} {user_agent.browser.version_string}",
            }
        except Exception as e:
            logger.error(f"获取用户代理信息失败: {str(e)}")
            return {"os": "", "browser": ""}

    @staticmethod
    def import_modules(modules: List[str], desc: str, **kwargs) -> None:
        """动态导入并执行模块

        Args:
            modules: 模块列表
            desc: 描述信息
            **kwargs: 其他参数
        """
        for module in modules:
            if not module:
                continue

            try:
                module_path = module[0 : module.rindex(".")]
                module_name = module[module.rindex(".") + 1 :]
                module_pag = importlib.import_module(module_path)
                getattr(module_pag, module_name)(**kwargs)
            except ModuleNotFoundError as e:
                logger.error(f"导入{desc}失败，模块：{module}，错误：{str(e)}")
            except AttributeError as e:
                logger.error(f"导入{desc}失败，方法：{module}，错误：{str(e)}")

    @staticmethod
    async def import_modules_async(modules: List[str], desc: str, **kwargs) -> None:
        """动态导入并执行异步模块

        Args:
            modules: 模块列表
            desc: 描述信息
            **kwargs: 其他参数
        """
        for module in modules:
            if not module:
                continue

            try:
                module_path = module[0 : module.rindex(".")]
                module_name = module[module.rindex(".") + 1 :]
                module_pag = importlib.import_module(module_path)
                await getattr(module_pag, module_name)(**kwargs)
            except ModuleNotFoundError as e:
                logger.error(f"导入{desc}失败，模块：{module}，错误：{str(e)}")
            except AttributeError as e:
                logger.error(f"导入{desc}失败，方法：{module}，错误：{str(e)}")

    @staticmethod
    def list_subpackages(package_name: str) -> List[str]:
        """列出包的所有子包

        Args:
            package_name: 包名

        Returns:
            List[str]: 子包列表
        """
        return [name for _, name, is_pkg in pkgutil.iter_modules(path=[package_name]) if is_pkg]

    @staticmethod
    def random_int(length: int) -> int:
        """生成指定长度的随机整数

        Args:
            length: 长度

        Returns:
            int: 随机整数
        """
        return random.randint(0, 10**length - 1)

    @staticmethod
    def make_uuid() -> str:
        """生成UUID

        Returns:
            str: UUID字符串
        """
        return uuid.uuid4().hex

    @staticmethod
    def make_md5(data: str) -> str:
        """生成MD5哈希值

        Args:
            data: 输入数据

        Returns:
            str: MD5哈希值
        """
        return hashlib.md5(data.encode("utf-8")).hexdigest()

    @staticmethod
    def make_token() -> str:
        """生成唯一Token

        Returns:
            str: Token字符串
        """
        ms = int(time.time() * 1000)
        token = ToolsUtils.make_md5(f"{ToolsUtils.make_uuid()}{ms}{ToolsUtils.random_string(8)}")
        token_secret = f"{token}{settings.SECRET_KEY}"
        return f"{ToolsUtils.make_md5(token_secret)}{ToolsUtils.random_string(6)}"

    @staticmethod
    def underscore_to_camelcase(s: str) -> str:
        """下划线命名转驼峰命名

        Args:
            s: 下划线命名字符串

        Returns:
            str: 驼峰命名字符串
        """
        return "".join(word.capitalize() if i != 0 else word for i, word in enumerate(s.split("_")))

    @staticmethod
    def convert_keys_to_camelcase(
        d: DictT,
    ) -> dict[str, dict[str, Any] | Any] | list[dict[str, Any] | Any] | dict | list:
        """将字典的键转换为驼峰命名

        Args:
            d: 输入字典

        Returns:
            DictT: 转换后的字典
        """
        if isinstance(d, dict):
            return {
                ToolsUtils.underscore_to_camelcase(k): ToolsUtils.convert_keys_to_camelcase(v) for k, v in d.items()
            }
        elif isinstance(d, list):
            return [ToolsUtils.convert_keys_to_camelcase(item) for item in d]
        return d

    @staticmethod
    def to_snake_case(s: str) -> str:
        """将字符串转换为蛇形命名

        Args:
            s: 输入字符串

        Returns:
            str: 蛇形命名字符串
        """
        return SNAKE_CASE_PATTERN.sub(r"\1_\2", s).lower()

    @staticmethod
    def convert_keys_to_snake_case(d: DictT) -> list[dict | Any] | dict[str, dict[str, Any] | Any] | dict | list:
        """将字典的键转换为蛇形命名

        Args:
            d: 输入字典

        Returns:
            DictT: 转换后的字典
        """
        if isinstance(d, dict):
            return {ToolsUtils.to_snake_case(k): ToolsUtils.convert_keys_to_snake_case(v) for k, v in d.items()}
        elif isinstance(d, list):
            return [ToolsUtils.convert_keys_to_snake_case(i) if isinstance(i, dict) else i for i in d]
        return d

    @staticmethod
    @lru_cache(maxsize=1024)
    def get_location_offline(ip: str) -> Optional[Dict[str, Any]]:
        """离线IP地址解析

        Args:
            ip: IP地址

        Returns:
            Optional[Dict[str, Any]]: 地理位置信息
        """
        if ToolsUtils.is_private_ip(ip):
            return {
                "country": "内网",
                "district": "内网",
                "province": "内网",
                "city": "内网",
                "continent": "内网",
                "isp": "内网",
                "latitude": None,
                "longitude": None,
                "area_code": None,
                "city_code": None,
            }

        try:
            cb = XdbSearcher.loadContentFromFile(settings.system.IP_DB_PATH)
            searcher = XdbSearcher(contentBuff=cb)
            result = searcher.search(ip)
            data = result.split("|")
            searcher.close()

            return {
                "country": data[0] if data[0] != "0" else None,
                "district": data[1] if data[1] != "0" else None,
                "province": data[2] if data[2] != "0" else None,
                "city": data[3] if data[3] != "0" else None,
                "continent": None,
                "isp": data[4] if data[4] != "0" else None,
                "latitude": None,
                "longitude": None,
                "area_code": None,
                "city_code": None,
            }
        except Exception as e:
            logger.error(f"离线IP解析失败: {str(e)}")
            return None

    @staticmethod
    def bytes2human(n: int, format_str: str = "%(value).1f%(symbol)s") -> str:
        """将字节数转换为人类可读格式

        Args:
            n: 字节数
            format_str: 格式化字符串

        Returns:
            str: 格式化后的字符串
        """
        symbols = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB")
        prefix = {s: 1 << (i + 1) * 10 for i, s in enumerate(symbols[1:])}

        for symbol in reversed(symbols[1:]):
            if n >= prefix[symbol]:
                value = float(n) / prefix[symbol]
                return format_str % locals()
        return format_str % dict(symbol=symbols[0], value=n)

    @staticmethod
    def human2bytes(s: str) -> int:
        """将人类可读格式转换为字节数

        Args:
            s: 人类可读格式的字符串

        Returns:
            int: 字节数

        Raises:
            ValueError: 无法解析的格式
        """
        init = s
        num = ""
        while s and (s[0:1].isdigit() or s[0:1] == "."):
            num += s[0]
            s = s[1:]

        num = float(num)
        letter = s.strip()

        for name, sset in SYMBOLS.items():
            if letter in sset:
                break
        else:
            if letter == "k":
                sset = SYMBOLS["customary"]
                letter = letter.upper()
            else:
                raise ValueError(f"无法解析格式: {init}")

        prefix = {sset[0]: 1}
        for i, s in enumerate(sset[1:]):
            prefix[s] = 1 << (i + 1) * 10

        return int(num * prefix[letter])

    @staticmethod
    def desensitize(data: DictT) -> DictT:
        """数据脱敏处理

        Args:
            data: 需要脱敏的数据

        Returns:
            DictT: 脱敏后的数据
        """
        print("data: ", data)
        if not data:
            return data

        for k, v in data.items():
            if isinstance(v, dict):
                data[k] = ToolsUtils.desensitize(v)
            elif k in settings.DESENSITIZE_FIELDS:
                data[k] = "*****"

        return data
